/*
 * 文件名：DataFileBackupService.java
 * 版权：亚信联创版权所有
 * 描述：TODO
 * 修改人：xuwei3
 * 修改时间：2014-11-14
 * 修改内容：TODO
 */
package com.ailk.bigdata.etl.realstream.server.service.impl;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.RandomAccessFile;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.common.server.date.DateUtil;
import com.ailk.bigdata.etl.common.server.io.FilenameUtil;
import com.ailk.bigdata.etl.common.server.io.IOUtil;
import com.ailk.bigdata.etl.common.server.thread.ThreadPool;
import com.ailk.bigdata.etl.realstream.server.model.RealDataDispatcher;
import com.ailk.bigdata.etl.realstream.server.model.RealStreamDataPack;

/**
 * @description 接收方接收未数据超时，写入文件缓存
 * @author [xuwei3]
 * @version [版本号,2014-11-14]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public class DataFileBackupService
{
	private static Logger logger = LoggerFactory.getLogger(DataFileBackupService.class);
	public static ConcurrentHashMap<Long, DataFileBackupService> WRITE_BACK_FILE_CAC = new ConcurrentHashMap<Long, DataFileBackupService>();
	private static String THREAD_BUFFER_POOL_NAME = "real_stream_buffer_pool";
	private static ThreadPool sendThreadPool = ThreadPool.getTheadPool(THREAD_BUFFER_POOL_NAME);
	private Semaphore writeReadLock = new Semaphore(1);
	private RealDataDispatcher dataDispatcher;
	private Long unitCode;
	private volatile boolean isWriteable = true;
	private String currentFileName;
	private ObjectInputStream inStream;
	private ObjectOutputStream outStream;

	public DataFileBackupService(RealDataDispatcher dataDispatcher)
	{
		this.dataDispatcher = dataDispatcher;
	}

	public static synchronized void backToFile(RealDataDispatcher dataDispatcher)
	{
		DataFileBackupService backupService = null;
		Long onlyUnitCode = dataDispatcher.getRealStreamUnit().getOnlyUnitCode();
		if (WRITE_BACK_FILE_CAC.containsKey(onlyUnitCode))
		{
			backupService = WRITE_BACK_FILE_CAC.get(onlyUnitCode);
		}
		else
		{
			backupService = new DataFileBackupService(dataDispatcher);
			WRITE_BACK_FILE_CAC.put(onlyUnitCode, backupService);
		}
		backupService.setWriteable(true);
		backupService.setUnitCode(onlyUnitCode);
		backupService.processSerial();
	}

	public static synchronized void getBufferedPack(RealDataDispatcher dataDispatcher)
	{
		Long onlyUnitCode = dataDispatcher.getRealStreamUnit().getOnlyUnitCode();
		if (WRITE_BACK_FILE_CAC.containsKey(onlyUnitCode))
		{
			DataFileBackupService backupService = WRITE_BACK_FILE_CAC.get(onlyUnitCode);
			backupService.setWriteable(false);
			backupService.processUnSerial();
		}
	}

	/**
	 * 
	 * @description 序列化
	 */
	private void processSerial()
	{
		sendThreadPool.runTask(new Runnable()
		{
			@Override
			public void run()
			{
				try
				{
					writeReadLock.acquire();
					outStream = null;
					logger.info("序列化线程开始执行.........only_unit_code[" + unitCode + "].......");
					RealStreamDataPack dataPack = null;
					while (null != (dataPack = dataDispatcher.pollMessage()) && isWriteable)
					{
						serialToFile(dataPack);
					}
					logger.info("发送channel开启，序列化线程执行结束.........only_unit_code[" + unitCode + "].......");
				}
				catch (InterruptedException e)
				{
					logger.error("", e);
				}
				finally
				{
					if (null != outStream)
					{
						try
						{
							outStream.flush();
						}
						catch (IOException e)
						{
							logger.error("", e);
						}
						IOUtil.closeQuietly(outStream);
					}
					writeReadLock.release();
				}
			}
		});
	}

	/**
	 * 
	 * @description 反序列化
	 */
	private void processUnSerial()
	{
		sendThreadPool.runTask(new Runnable()
		{
			@Override
			public void run()
			{
				try
				{
					writeReadLock.acquire();
					inStream = null;
					logger.info("反序列化开始.........only_unit_code[" + unitCode + "].......");
					UnserialToQueue();
					currentFileName = null;
					logger.info("反序列化成功.........only_unit_code[" + unitCode + "].......");
				}
				catch (InterruptedException e)
				{
					logger.error("", e);
				}
				finally
				{
					writeReadLock.release();
				}
			}
		});
	}

	private void serialToFile(RealStreamDataPack dataPack)
	{
		try
		{
			if (null == outStream)
			{
				if (StringUtils.isEmpty(currentFileName))
				{
					currentFileName = "buffer_" + unitCode + "_" + DateUtil.getTimeStamp() + FilenameUtil.EXTENSION_SEPARATOR + "dat";
				}
				File file = new File(currentFileName);
				logger.info("currentFileName------>fileName[" + currentFileName + "]path[" + file.getAbsolutePath() + "]");
				outStream = new ObjectOutputStream(new BufferedOutputStream(new FileOutputStream(new RandomAccessFile(file, "rw").getFD())));
			}
			outStream.writeObject(dataPack);
		}
		catch (FileNotFoundException e)
		{
			logger.error("序列化失败", e);
		}
		catch (IOException e)
		{
			logger.error("序列化失败", e);
		}
	}

	private void UnserialToQueue()
	{
		try
		{
			if (StringUtils.isNotEmpty(currentFileName))
			{
				if (null == inStream)
				{
					inStream = new ObjectInputStream(new BufferedInputStream(new FileInputStream(new RandomAccessFile(currentFileName, "r").getFD())));
				}
				RealStreamDataPack dataPack = null;
				while (null != (dataPack = (RealStreamDataPack) inStream.readObject()))
				{
					dataDispatcher.addMessage(dataPack);
				}
			}
		}
		catch (EOFException e)
		{
			logger.info("文件读取结束，反序列化结束.........");
		}
		catch (FileNotFoundException e)
		{
			logger.error("反序列化失败", e);
		}
		catch (IOException e)
		{
			logger.error("反序列化失败", e);
		}
		catch (ClassNotFoundException e)
		{
			logger.error("反序列化失败", e);
		}
		finally
		{
			if (StringUtils.isNotEmpty(currentFileName))
			{
				File file = new File(currentFileName);
				file.delete();
			}
			IOUtil.closeQuietly(inStream);
		}
	}

	public boolean isWriteable()
	{
		return isWriteable;
	}

	public void setWriteable(boolean isWriteable)
	{
		this.isWriteable = isWriteable;
	}

	public String getCurrentFileName()
	{
		return currentFileName;
	}

	public void setCurrentFileName(String currentFileName)
	{
		this.currentFileName = currentFileName;
	}

	public Long getUnitCode()
	{
		return unitCode;
	}

	public void setUnitCode(Long unitCode)
	{
		this.unitCode = unitCode;
	}

}
